package com.test.study.ToolsUtils.strom.workdsum;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.esotericsoftware.minlog.Log;

public class DataBolt extends BaseRichBolt {

	private static final long serialVersionUID = -670790067686165101L;
	@SuppressWarnings({ "rawtypes", "unused" })
	private Map stormConf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	private OutputCollector collector;

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.stormConf = stormConf;
		this.context = context;
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		try {
			String stringByField = input.getStringByField("line");
			if (!stringByField.isEmpty()) {
				String[] splits = stringByField.split("\t");
				if (splits != null && splits.length > 0) {
					for (String split : splits) {
						this.collector.emit(input,new Values(split));//input 这个时候相当于把word所在的tuple和上一级tuple绑定到一块立刻了,也就是说 word 这个tuple是上一级tuple的衍生
					}
				}
			}
			this.collector.ack(input);
		} catch (Exception e) {
			Log.error(e.getMessage());
			this.collector.fail(input);
		}
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));

	}

}
